#!/usr/bin/env python3
# -*- coding: utf-8 -*-
########################################################################
#
# Copyright (c) 2023 Zhiyong Lin. All Rights Reserved
#
########################################################################
'''
@File    :   push_data.py
@Time    :   2025/05/10 19:42:50
@Author  :   lzy2lcy 
@Version :   1.0
@Desc    :   
'''

import json
import time
import random
from confluent_kafka import Producer

# 配置 Kafka 生产者
conf = {
    'bootstrap.servers': '192.168.124.16:19092,192.168.124.16:19093,192.168.124.16:19094',
    'client.id': 'json-producer'
}
print(conf)
# 初始化 Kafka 生产者
producer = Producer(conf)

# 生成模拟数据
def generate_data():
    names = ['Alice', 'Bob', 'Charlie', 'David', 'Eve']
    ip_parts = [str(random.randint(0, 255)) for _ in range(4)]
    return {
        'time': int(time.time()),
        'name': random.choice(names),
        'ip': '.'.join(ip_parts)
    }

# 消息回调函数（可选）
def delivery_report(err, msg):
    if err:
        print(f'Message delivery failed: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}]')

# 主循环：每秒发送一次数据
try:
    while True:
        data = generate_data()
        json_data = json.dumps(data)

        # 发送数据到 Kafka 主题（例如 "test-topic"）
        producer.produce(
            'test',
            key='key',
            value=json_data,
            callback=delivery_report
        )

        # 刷新缓冲区，确保消息及时发送
        producer.poll(0)
        print("push to kafka done. data:" + str(data))

        time.sleep(10)

except KeyboardInterrupt:
    print("Producer interrupted.")

finally:
    # 等待所有消息发送完成
    producer.flush()
    print("Producer closed.")